package com.powerbank.common.mq.producer;

import com.powerbank.common.event.BaseEvent;
import com.powerbank.common.mq.EventPublisher;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * 消息生产者工具类
 * 提供消息发送的统一接口
 */
@Slf4j
@Component
public class MessageProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private EventPublisher eventPublisher;

    /**
     * 同步发送消息
     */
    public SendResult sendSync(String topic, String tag, Object payload) {
        try {
            String destination = topic + ":" + tag;
            log.info("同步发送消息: destination={}, payload={}", destination, payload);
            
            SendResult result = rocketMQTemplate.syncSend(destination, payload);
            log.info("消息发送成功: msgId={}, queueId={}", result.getMsgId(), result.getMessageQueue().getQueueId());
            
            return result;
        } catch (Exception e) {
            log.error("同步发送消息失败: topic={}, tag={}", topic, tag, e);
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 异步发送消息
     */
    public void sendAsync(String topic, String tag, Object payload, SendCallback callback) {
        try {
            String destination = topic + ":" + tag;
            log.info("异步发送消息: destination={}, payload={}", destination, payload);
            
            rocketMQTemplate.asyncSend(destination, payload, callback);
        } catch (Exception e) {
            log.error("异步发送消息失败: topic={}, tag={}", topic, tag, e);
            if (callback != null) {
                callback.onException(e);
            }
        }
    }

    /**
     * 单向发送消息（不关心发送结果）
     */
    public void sendOneWay(String topic, String tag, Object payload) {
        try {
            String destination = topic + ":" + tag;
            log.info("单向发送消息: destination={}, payload={}", destination, payload);
            
            rocketMQTemplate.sendOneWay(destination, payload);
        } catch (Exception e) {
            log.error("单向发送消息失败: topic={}, tag={}", topic, tag, e);
        }
    }

    /**
     * 发送延时消息
     */
    public SendResult sendDelaySync(String topic, String tag, Object payload, int delayLevel) {
        try {
            String destination = topic + ":" + tag;
            Message<Object> message = MessageBuilder.withPayload(payload).build();
            
            log.info("发送延时消息: destination={}, delayLevel={}, payload={}", 
                    destination, delayLevel, payload);
            
            SendResult result = rocketMQTemplate.syncSend(destination, message, 3000, delayLevel);
            log.info("延时消息发送成功: msgId={}, queueId={}", result.getMsgId(), result.getMessageQueue().getQueueId());
            
            return result;
        } catch (Exception e) {
            log.error("发送延时消息失败: topic={}, tag={}, delayLevel={}", topic, tag, delayLevel, e);
            throw new RuntimeException("延时消息发送失败", e);
        }
    }

    /**
     * 发送顺序消息
     */
    public SendResult sendOrderly(String topic, String tag, Object payload, String hashKey) {
        try {
            String destination = topic + ":" + tag;
            log.info("发送顺序消息: destination={}, hashKey={}, payload={}", destination, hashKey, payload);
            
            SendResult result = rocketMQTemplate.syncSendOrderly(destination, payload, hashKey);
            log.info("顺序消息发送成功: msgId={}, queueId={}", result.getMsgId(), result.getMessageQueue().getQueueId());
            
            return result;
        } catch (Exception e) {
            log.error("发送顺序消息失败: topic={}, tag={}, hashKey={}", topic, tag, hashKey, e);
            throw new RuntimeException("顺序消息发送失败", e);
        }
    }

    /**
     * 发送事务消息
     */
    public void sendInTransaction(String topic, String tag, Object payload, Object arg) {
        try {
            String destination = topic + ":" + tag;
            Message<Object> message = MessageBuilder.withPayload(payload).build();
            
            log.info("发送事务消息: destination={}, payload={}, arg={}", destination, payload, arg);
            
            rocketMQTemplate.sendMessageInTransaction(destination, message, arg);
        } catch (Exception e) {
            log.error("发送事务消息失败: topic={}, tag={}", topic, tag, e);
            throw new RuntimeException("事务消息发送失败", e);
        }
    }

    /**
     * 发布业务事件（使用EventPublisher）
     */
    public void publishEvent(BaseEvent event) {
        try {
            String eventType = event.getEventType();
            if (eventType.startsWith("USER_")) {
                eventPublisher.publishUserEvent(event);
            } else if (eventType.startsWith("PAYMENT_")) {
                eventPublisher.publishPaymentEvent(event);
            } else if (eventType.startsWith("ORDER_")) {
                eventPublisher.publishOrderEvent(event);
            } else if (eventType.startsWith("DEVICE_")) {
                eventPublisher.publishDeviceEvent(event);
            } else if (eventType.startsWith("FINANCE_")) {
                eventPublisher.publishFinanceEvent(event);
            } else {
                eventPublisher.publishNotificationEvent(event);
            }
        } catch (Exception e) {
            log.error("发布业务事件失败: eventType={}, eventId={}", 
                     event.getEventType(), event.getEventId(), e);
            throw new RuntimeException("业务事件发布失败", e);
        }
    }

    /**
     * 创建默认的异步回调
     */
    public static SendCallback createDefaultCallback(String operation) {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("{}异步消息发送成功: msgId={}, queueId={}", 
                        operation, sendResult.getMsgId(), sendResult.getMessageQueue().getQueueId());
            }

            @Override
            public void onException(Throwable e) {
                log.error("{}异步消息发送失败", operation, e);
            }
        };
    }
}